import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class Demo07_tempWarning {
    public static void tempWarning(DataStream<Demo02_sensorReading> sensor){

        sensor
            .keyBy(Demo02_sensorReading::getSensorId)
            .process(
                    new ProcessTemperatureWarning()
            )
            .print();

    }

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> lineDataStream = Demo02_source.source_socket(env,9999);

        DataStream<Demo02_sensorReading> sensorDataStream = Demo02_transformation.transformation_simple(lineDataStream);

        tempWarning(sensorDataStream);

        env.execute("warning");

    }

}

/**
 * 自定义类，实现底层API的方法
 */
class ProcessTemperatureWarning extends KeyedProcessFunction<String,Demo02_sensorReading,String> {

    //声明一个 value类型 Double 泛型的状态，用来存储每一条数据的上一条温度值
    ValueState<Double> lastTempState;
    //声明一个 value类型 Long 泛型的状态，使用时间戳的形式存储此时此刻是否存在定时器
    ValueState<Long> timerTSState;
    //声明一个 List类型 Double 泛型的状态，用来存储当前10s内连续上升的所有温度值
    ListState<Double> allTempASCState;


    @Override
    public void open(Configuration parameters) throws Exception {

        //全局执行一次，用来定义三个状态
        lastTempState = getRuntimeContext().getState(
                new ValueStateDescriptor<Double>("lastTemp",Double.class)
        );

        timerTSState = getRuntimeContext().getState(
                new ValueStateDescriptor<Long>("timerTS",Long.class)
        );

        allTempASCState = getRuntimeContext().getListState(
                new ListStateDescriptor<Double>("allTempASC",Double.class)
        );

    }

    /**
     * 处理每条数据
     * @param value The input value.
     */
    @Override
    public void processElement(Demo02_sensorReading value, KeyedProcessFunction<String, Demo02_sensorReading, String>.Context ctx, Collector<String> out) throws Exception {

        //获取上一条温度的值
        double lastTemp = lastTempState.value()==null?0:lastTempState.value();
        //获取当前条数据的温度值
        double currentTemp = value.getTemperature();

        //获取定时器的状态，判断此时是否已经存在定时器了
        long timer = timerTSState.value()==null?0:timerTSState.value();

        //任何条件都需要把当前温度值更新到上一条温度值的状态中
        lastTempState.update(currentTemp);

        //第一条数据和正常执行onTimer后需要执行的内容
        if(timer==0){

            System.out.println("此时没有定时器，需要新建定时器");

            //1.先获取定时器对象
            TimerService timerService = ctx.timerService();
            //2.获取当前定时器的时间=当前条数据被处理的时间+往后延迟的10s钟
            long timerTS = timerService.currentProcessingTime()+10*1000;
            //3.注册定时器
            timerService.registerProcessingTimeTimer(timerTS);

            //4.用当前的定时器时间戳更新定时器状态
            timerTSState.update(timerTS);

            System.out.println(timerTS+"定时器已经被创建,start ====="+System.currentTimeMillis()/1000+"=====");

        }else if(timer>0 && currentTemp<=lastTemp){
            //在有定时器的情况下，温度值出现了下降趋势时执行

            System.out.println("此时有定时器，但是温度下降了，删除"+timer+"定时");

            //1.先获取定时器对象,为了在当前定时器内，上升趋势的情况不创建此对象，这里没有提出去
            TimerService timerService = ctx.timerService();
            //2.删除原有定时器
            timerService.deleteProcessingTimeTimer(timer);

            //3.获取新建定时器的时间=当前条数据被处理的时间+往后延迟的10s钟
            long timerTS = timerService.currentProcessingTime()+10*1000;

            System.out.println(timer+"定时器已经删除完毕，同时创建"+timerTS+"定时器," +
                    "start ====="+System.currentTimeMillis()/1000+"=====");

            //4.清空当前定时器内所有温度值的列表
            allTempASCState.clear();

            //5.注册定时器
            timerService.registerProcessingTimeTimer(timerTS);

            //6.用当前的定时器时间戳更新定时器状态
            timerTSState.update(timerTS);

            System.out.println(timerTS+"定时器已经被创建");


        }else if(timer>0 && currentTemp>lastTemp){
            //在有定时器的情况下，温度值保持上升趋势时，把当前条温度值加到列表中
            allTempASCState.add(currentTemp);
        }


    }

    @Override
    public void onTimer(long timestamp, KeyedProcessFunction<String, Demo02_sensorReading, String>.OnTimerContext ctx, Collector<String> out) throws Exception {

        System.out.println("执行定时器的时间 ====="+System.currentTimeMillis()/1000+"=====");

        //1.封装报警信息
        String warningMessage = "温度已经连续10s钟上升了，请注意！请注意！请注意！温度值为："+allTempASCState.get();
        //2.发送报警信息
        out.collect(warningMessage);
        //3.清空定时器状态内容，以空定时器的状态迎接下一条数据
        timerTSState.clear();
        //4.清空当前定时器内所有温度值的列表
        allTempASCState.clear();
    }
}